package com.qiangesoft.mqtt.subscriber;

import com.qiangesoft.mqtt.config.MqttProperty;
import com.qiangesoft.mqtt.constant.MqttConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.Objects;

/**
 * mqtt消费者
 *
 * @author qiangesoft
 * @date 2024-04-24
 */
@Slf4j
@Configuration
public class MqttSubscriberConfig {

    @Autowired
    private MqttProperty mqttProperty;
    @Autowired
    private MqttPahoClientFactory mqttPahoClientFactory;

    /**
     * 消息订阅通道
     *
     * @return
     */
    @Bean(name = MqttConstant.INBOUND_CHANNEL)
    public MessageChannel inboundChannel() {
        return new DirectChannel();
    }

    /**
     * 消息订阅通道绑定
     *
     * @return
     */
    @Bean
    public MessageProducer mqttInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperty.getSubscriber().getClientId(),
                mqttPahoClientFactory, mqttProperty.getSubscriber().getTopic());
        adapter.setCompletionTimeout(mqttProperty.getCompletionTimeout());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(mqttProperty.getSubscriber().getQos());
        adapter.setOutputChannel(inboundChannel());
        return adapter;
    }

    /**
     * 消息订阅
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConstant.INBOUND_CHANNEL)
    public MessageHandler messageHandler() {
        return message -> {
            try {
                String topic = Objects.requireNonNull(message.getHeaders().get("mqtt_receivedTopic")).toString();
                log.info("订阅主题为: {}", topic);
                String payload = message.getPayload().toString();
                log.info("订阅接收到消息：{}", payload);
            } catch (Exception e) {
                e.printStackTrace();
            }
        };
    }

}


